package com.csw.flink.source

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.functions.source.SourceFunction

object Demo02SourceFunction {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 使用自定义source
      *
      */

    val ds: DataStream[Int] = env.addSource(new MySource)

    ds.print()

    env.execute()

  }

  /**
    * 自定义source，实现SourceFunction接口
    */

  class MySource extends SourceFunction[Int] {
    /**
      * run  用于生产数据，把数据发送到下游算子
      *
      * 任务开始的时候运行一次
      */
    override def run(sourceContext: SourceFunction.SourceContext[Int]): Unit = {

      var i = 0

      while (true) {
        i = i + 1

        //将数据发送到下游算子
        sourceContext.collect(i)

        //睡眠一秒钟
        Thread.sleep(1000)
      }
    }

    /**
      * 任务取消的时候会执行的一个方法，  一般用于回收资源
      *
      * 任务取消的时候运行一次
      */
    override def cancel(): Unit = {

    }
  }

}
